package com.sjc.lesson02.state.keyedState;

import com.sjc.lesson02.state.keyedState.function.ContainsValueFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 需求： 当接收到的相同的key的元素的个数等于3个，就计算这些元素的value的平均值。
 * 计算 keyed stream中每3个元素的value的平均值
 *
 * 每3个数求一次
 */
public class TestKeyedStateMain {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<Tuple2<Long, Long>> dataStreamSource =
                env.fromElements(Tuple2.of(1L, 3L), Tuple2.of(1L, 5L), Tuple2.of(1L, 7L),
                        Tuple2.of(2L, 4L), Tuple2.of(2L, 2L), Tuple2.of(2L, 5L)
                        );
        dataStreamSource.keyBy(0)
//                .flatMap(new CountWindowAverageWithValueState())
//                .flatMap(new CountWindowAverageWithListState())
//                .flatMap(new CountWindowAverageWithMapState())
//                .flatMap(new SumFunction())
                .flatMap(new ContainsValueFunction())
                .print();
        env.execute("TestKeyedStateMain");
    }
}
